package com.etc

import java.sql.Date
import java.text.SimpleDateFormat

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession


/**
  * Movie数据集 数据集字段进行分割
  *
  * 1^                           电影ID
  * Toy Story (1995)^            电影的名字
  * In the highlands ....^       电影的描述
  * 85^                          电影的时长
  * August 26, 1997^              电影的发行日期
  * 1995^                         电影的拍摄日期
  * English ^                     电影的语言
  * Action|Drama|Romance|War ^    电影的类型
  * Liam Neeson|Jessica Lange...  电影的演员
  * Michael Caton-Jones           电影的导演
  * tag1|tag2|tag3|....           电影的Tag
  */
case class Movie(val mid: Int, val name: String, val descri: String, val timelong: String, val issue: String,
                 val shoot: String, val language: String, val genres: String, val actors: String, val directives: String)


/**
  * Rating数据集：用户对电影的评分数据集
  * 1                    用户ID
  * 31                   电影ID
  * 2.5                  用户对于电影的评分
  * 1260759144           用户对于电影的评分的时间
  */
case class Rating(val uid: Int, val mid: Int, val score: Double, val timestamp: Int)


/**
  * Mongodb的连接配置
  *
  * @param url
  * @param db
  */
case class MongodbConfig(val url: String, val db: String)




/**
  * 推荐对象
  * @param rid   推荐的Movie的mid
  * @param r     Movie的评分
  */
case class Recommendation(rid:Int, r:Double)


/**
  * 电影类别的推荐
  * @param genres   电影的类别
  * @param recs     top10的电影的集合
  */
case class GenresRecommendation(genres:String, recs:Seq[Recommendation])


/**
  * 做分析
  */
object StatisticRecommender {

 //从mongodb读取的数据
  val MONGODB_RATING_COLLECTION = "Rating"
  val MONGODB_MOVIE_COLLECTION = "Movie"


    //处理后数据后的覆盖表
    val RATE_MORE_MOVIE = "RateMoreMovies"
    val RATE_MORE_RECENTLY_MOVIES ="RateMoreRecentlyMovies"
    val AVERAGE_MOVIES = "AverageMovies"
    val GENRES_TOP_MOVIES = "GenresTopMovies"


  def main(args: Array[String]): Unit = {

    val config = Map(
      "spark.name" -> "StatisticsRecommender",
      "spark.cores" -> "local[*]",
      "mongodb.uri" -> "mongodb://localhost:27017/movies",
      "mongodb.db" -> "movies"
    )

    //需要创建一个Sparkconf 配置
    val conf = new SparkConf().setMaster(config("spark.cores")).setAppName(config("spark.name"))
    //创建一个sparksession
    val spark = SparkSession.builder().config(conf).getOrCreate()

    import  spark.implicits._

    //创建一个连接monggodb配置
    val mongodbconfig = MongodbConfig(config("mongodb.uri"),config("mongodb.db"))

//    从mongodb 把评分数据加载至Dataframe
    val ratingDF = spark.read
      .option("uri",mongodbconfig.url)
      .option("collection",MONGODB_RATING_COLLECTION)     //表名
      .format("com.mongodb.spark.sql")
      .load()
      .as[Rating]
      .toDF

    val movieDF = spark.read
      .option("uri",mongodbconfig.url)
      .option("collection",MONGODB_MOVIE_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Movie]
      .toDF

    //######################################  -->统计每个电影评分的次数   开始<--#################################################

   //创建一个rating的临时表才可以进行sql查询
    ratingDF.createOrReplaceTempView("ratings")
   //统计所有历史数据中每个电影的评分次数总数(需要电影mid，count)
    val rateMoreMoviesDF = spark.sql("select mid ,count(mid) as count from ratings group  by mid")

//    rateMoreMoviesDF
//      .write
//      .option("uri",mongodbconfig.url)
//      .option("collection",RATE_MORE_MOVIE)
//      .mode("overwrite")
//      .format("com.mongodb.spark.sql")
//      .save() //保存


    //######################################  -->统计每个电影评分的次数   结束<--#################################################


    //######################################  -->统计已月为单位每个电影的评分数  开始<--#################################################

     //创建一个日期格式转化工具
    val simpleDataFormat = new SimpleDateFormat("yyyyMM")
    //注册一个UDF函数，用于将timestamp装换成年月格式   1260759144000  => 201605
    spark.udf.register("changeDate",(x:Int) => simpleDataFormat.format(new Date(x * 1000L)).toInt)

    //将原来的Rating数据集中的时间转换成年月的格式(查询的是mid)
    val ratingOfYeahMouth = spark.sql("select mid, score,changeDate(timestamp) as yeshmouth from ratings")

    // 把新的数据制作一个表
    ratingOfYeahMouth.createOrReplaceTempView("ratingOfMouth")
    val rateMoreRecentlyMovies = spark.sql("select mid, count(mid) count,yeshmouth from ratingOfMouth group by yeshmouth,mid")
//    rateMoreRecentlyMovies
//      .write
//      .option("uri",mongodbconfig.url)
//      .option("collection",RATE_MORE_RECENTLY_MOVIES)
//      .mode("overwrite")
//      .format("com.mongodb.spark.sql")
//      .save()    //保存

    //######################################  -->统计已月为单位每个电影的评分数  结束<--#################################################


    //######################################  -->统计每个电影的平均评分  开始<--#################################################
     val avgrageMoviesDF = spark.sql("select mid, avg(score) as avg from ratings group by mid")
//      avgrageMoviesDF
//          .write
//          .option("uri",mongodbconfig.url)
//          .option("collection",AVERAGE_MOVIES)
//          .mode("overwrite")
//          .format("com.mongodb.spark.sql")
//          .save()     //保存

    //######################################  -->统计每个电影的平均评分  结束<--#################################################


    //######################################  -->统计每种电影类型中评分最高的10个电影  开始<--#################################################


    //需要用left join(全部的点击数据集合join平均评分数据集)，应为只需要  (有)  评分的电影数据集
    val movieWithScore = movieDF.join(avgrageMoviesDF	,Seq("mid","mid"))

    //所有的电影类别
    val genres = List("Action","Adventure","Animation","Comedy","Ccrime","Documentary","Drama","Family","Fantasy","Foreign","History","Horror","Music","Mystery"
      ,"Romance","Science","Tv","Thriller","War","Western")

    //将电影类别转换成RDD(转化RDD可以用 makeRDD  还可以用 parallelize)
    val genresRDD = spark.sparkContext.parallelize(genres)


    //计算电影类别top10
    val genrenTopMovies = genresRDD.cartesian(movieWithScore.rdd)  //将电影类别和电影数据进行笛卡尔积操作
      .filter{
      // 过滤掉电影的类别不匹配的电影
      case (genres,row) => row.getAs[String]("genres").toLowerCase.contains(genres.toLowerCase)
    }
      .map{
        // 将整个数据集的数据量减小，生成RDD[String,Iter[mid,avg]]
        case (genres,row) => {
          (genres,(row.getAs[Int]("mid"), row.getAs[Double]("avg")))
        }
      }.groupByKey()   //将genres数据集中的相同的聚集
      .map{
      // 通过评分的大小进行数据的排序，然后将数据映射为对象
      case (genres, items) => GenresRecommendation(genres,items.toList.sortWith(_._2 > _._2).take(10).map(item => Recommendation(item._1,item._2)))
    }.toDF()

    // 输出数据到MongoDB
//    genrenTopMovies
//      .write
//      .option("uri",mongodbconfig.url)
//      .option("collection",GENRES_TOP_MOVIES)
//      .mode("overwrite")
//      .format("com.mongodb.spark.sql")
//      .save()
    //######################################  -->统计每种电影类型中评分最高的10个电影  结束<--#################################################

    spark.stop()

  }
}
